package com.wuwangfu.join;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.join
 * @ClassName: TumbleWindowJoin
 * @Description:
 * @Version 1.0.0
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/joining/#tumbling-window-join
 *
 *
 * join 连接
 */
public class TumbleWindowJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        /**
         * 1000,A,111
         * 4999,A,222
         *
         * 6666,A,111
         * 9999,A,222
         */
        DataStreamSource<String> left = env.socketTextStream("localhost", 8888);
        /**
         * 3333,B,9999
         * 4999,B,8888
         *
         * 7777,A,a
         * 10000,B,b
         */
        DataStreamSource<String> right = env.socketTextStream("localhost", 9999);
        //提取 left 流中数据的EventTime
        SingleOutputStreamOperator<String> leftWithWatermark = left.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //提取 right 流中数据的EventTime
        SingleOutputStreamOperator<String> rightWithWatermark = right.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        // left -> tuple3
        SingleOutputStreamOperator<Tuple3<Long, String, String>> leftmaped = leftWithWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });

        // right -> tuple3
        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightmaped = rightWithWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });

        //调用join方法，并且在where方法和equalTo方法中分别指定两个流join的条件
        //第一个流（左流）和第二个流（右流）进行join，只有在同一个窗口内，并且join的条件相等
        leftmaped.join(rightmaped)
                //指定 join 的条件，第一个流的条件
                .where(new KeySelector<Tuple3<Long, String, String>, String>() {
                    @Override
                    public String getKey(Tuple3<Long, String, String> value) throws Exception {
                        //将 leftmaped 流中的 f1 作为 join 的key
                        return value.f1;
                    }
                })
                //第二个流的条件
                .equalTo(new KeySelector<Tuple3<Long, String, String>, String>() {
                    @Override
                    public String getKey(Tuple3<Long, String, String> value) throws Exception {
                        //将 rightmaped 流的 f1 作为 join 的key
                        return value.f1;
                    }
                })
                //按EventTime划分滚动窗口，窗口长度为5秒，底层对join的条件进行keyBy
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //在apply方法中传入自定义的 join 方法
                .apply(new JoinFunction<
                        //第一个流
                        Tuple3<Long, String, String>,
                        //第二个流
                        Tuple3<Long, String, String>,
                        //返回值
                        Tuple5<Long, Long, String, String, String>>() {
                    @Override
                    public Tuple5<Long, Long, String, String, String> join(Tuple3<Long, String, String> first, Tuple3<Long, String, String> second) throws Exception {
                        //如果进入到这个join方法中，就意味着：join的条件相等，在同一个窗口
                        return Tuple5.of(first.f0, second.f0, first.f1, first.f2, second.f2);
                    }
                }).print();


        env.execute();
    }

}
